package com.ehotting.mqbd.core.service.msg;

import com.ehotting.eaf.core.utils.AssertUtils;
import com.ehotting.eaf.redis.service.RedisService;
import com.ehotting.mqbd.api.MQDBConstants;
import com.ehotting.mqbd.api.bean.MsgStateBean;
import com.ehotting.mqbd.api.enums.MsgStateEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.TimeUnit;

import java.util.stream.Collectors;

@Component("msgStateService")
public class MsgStateService {

    private static final Logger logger = LoggerFactory.getLogger(MsgStateService.class);

    @Autowired
    private RedisService redisService;


    private TimeUnit TIME_UNIT = TimeUnit.DAYS;

    private long TIME_OUT = MQDBConstants.TIME_OUT;


    public void put(MsgStateBean msgState){
        logger.info("Put msg[agentId:{},msgId:{}]",msgState.getAgentId(),msgState.getMsgId());
        redisService.setCacheObject(MQDBConstants.QUEUE_MSG + msgState.getMsgId(),msgState,TIME_OUT,TIME_UNIT);
    }

    public void put(String msgId,String agentId,String serial){
        put(new MsgStateBean(agentId,msgId,serial));
    }

    public MsgStateBean getByMsgId(String msgId){
        return redisService.getCacheObject(MQDBConstants.QUEUE_MSG +msgId);
    }


    public void deleteByMsgId(String msgId){
        logger.info("Delete msg[msgId:{}]",msgId);
        redisService.delete(MQDBConstants.QUEUE_MSG+msgId);
    }

    public List<MsgStateBean> findAll(){
        List<MsgStateBean> msgStateList = new ArrayList<>();
        Collection<String> keys = redisService.keys(MQDBConstants.QUEUE_MSG + "*");
        for(String key : keys){
            MsgStateBean msgState = redisService.getCacheObject(key);
            if(msgState==null){
                continue;
            }
            msgStateList.add(msgState);
        }
        return msgStateList;
    }



    public List<MsgStateBean> findByModule(String module){
        AssertUtils.hasText(module,"module is null.");
        List<MsgStateBean> msgStateList = new ArrayList<>();
        Collection<String> keys = redisService.keys(MQDBConstants.QUEUE_MSG + "*");
        for(String key : keys){
            MsgStateBean msgState = redisService.getCacheObject(key);
            if(msgState==null){
                continue;
            }
            if(msgState.getAgentId().indexOf(module)>=0){
                msgStateList.add(msgState);
            }
        }
        if(msgStateList.size()>0){
            msgStateList = msgStateList.stream().sorted(Comparator.comparing(MsgStateBean::getDate).reversed()).collect(Collectors.toList());
        }
        return msgStateList;
    }

    public List<MsgStateBean> findByAgentId(String agentId){
        AssertUtils.hasText(agentId,"agentId is null.");
        List<MsgStateBean> msgStateList = new ArrayList<>();
        Collection<String> keys = redisService.keys(MQDBConstants.QUEUE_MSG + "*");
        for(String key : keys){
            MsgStateBean msgState = redisService.getCacheObject(key);
            if(msgState==null){
                continue;
            }
            if(msgState.getAgentId().equals(agentId)){
                msgStateList.add(msgState);
            }
        }
        if(msgStateList.size()>0){
            msgStateList = msgStateList.stream().sorted(Comparator.comparing(MsgStateBean::getDate).reversed()).collect(Collectors.toList());
        }
        return msgStateList;
    }
}